package com.tricycle.scada.mqtt.realtime;

import com.tricycle.scada.mqtt.model.RealtimeMessageModel;
import com.tricycle.scada.mqtt.model.RealtimeValueModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
@Component
public class RealtimeValueRedisHandler implements RealtimeValueHandler {

    private final ExecutorService executorService = Executors.newFixedThreadPool(10);

    private final RedisTemplate<String, String> redisStringTemplate;

    public RealtimeValueRedisHandler(RedisTemplate<String, String> redisStringTemplate) {
        this.redisStringTemplate = redisStringTemplate;
    }

    @Override
    public void handle(RealtimeMessageModel message) {


        executorService.submit(() -> {
            Map<String, Set<ZSetOperations.TypedTuple<String>>> recordsMap = new HashMap<>();
            for (RealtimeValueModel value : message.getRealtimeValues()) {

                ZSetOperations.TypedTuple<String> record = new DefaultTypedTuple<>(value.getTimestamp() + "::" + value.getValue().toString(), value.getTimestamp().doubleValue());
                if (!recordsMap.containsKey(value.getName())) {
                    recordsMap.put(value.getName(), new HashSet<>());
                }
                recordsMap.get(value.getName()).add(record);
                recordsMap.computeIfAbsent(value.getName(), (k) -> new HashSet<>()).add(record);
            }
            redisStringTemplate.opsForSet().add("tags", recordsMap.keySet().toArray(new String[0]));
            for (String key : recordsMap.keySet()) {
                try {
                    redisStringTemplate.opsForZSet().add(key, recordsMap.get(key));
                } catch (Exception e) {
                    logger.error("Insert to redis error: {}", e.getMessage());
                }
            }
        });

    }
}
